package com.meter.reading.listener;

import com.alibaba.fastjson.JSONObject;
import com.meter.common.utils.StringUtils;
import com.meter.netty.channel.NettyChannel;
import com.meter.reading.util.RedisRunnable;
import com.meter.redis.service.RedisService;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class TcpListener {

    @Autowired
    private RedisService redisService;

    @Autowired
    private NettyChannel nettyChannel;

    /**
     * 采集器批量采集
     * @param deviceCode
     * @param channel
     * @param message
     * @throws Exception
     */
    @RabbitListener(queues = {"device_queue"})
    @RabbitHandler
    public void deviceQueue(String deviceCode, Channel channel, Message message) throws Exception{
        try {
            String messageId = message.getMessageProperties().getHeaders().get("spring_returned_message_correlation").toString();
            if (redisService.hasKey("message:" + messageId)) {
                return;
            }
            redisService.set("message:" + messageId, messageId, 10);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

            //根据设备识别码查询通道号
            String channelId = redisService.hget("sys-netty", deviceCode);
            Thread daemonThread = new Thread(() -> {
                int count = 0;
                while (true) {
                    try {
                        if (redisService.hasKey("reading_meter:" + deviceCode)) {
                            if(count < 20) {
                                Thread.sleep(1000);
                                count ++;
                            } else {
                                count = 0;
                                redisService.del("reading_meter:" + deviceCode);
                            }
                        } else {
                            JSONObject readingJson = (JSONObject) redisService.leftPop("device:" + deviceCode);
                            if (StringUtils.isNull(readingJson)) {
                                redisService.del("readingThread:" + deviceCode);
                                break;
                            }
                            redisService.set("reading_meter:" + deviceCode, readingJson, 20);
                            //发送采集指令码
                            nettyChannel.writeToClient(channelId, readingJson.getString("receiveStr"), readingJson.getString("mark"));
                            count = 0;
                        }
                    } catch (InterruptedException e) {
                        log.error("运行异常：" + e.getMessage());
                        throw new RuntimeException(e);
                    }
                }
            });
            redisService.set("readingThread:" + deviceCode, daemonThread.getId());
            // 将线程设置为守护线程
            daemonThread.setDaemon(true);
            daemonThread.start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 热表数据单独采集
     * @param jsonObject
     * @param channel
     * @param message
     * @throws Exception
     */
    @RabbitListener(queues = {"job_queue"})
    @RabbitHandler
    public void jobQueue(JSONObject jsonObject, Channel channel, Message message) {
        try {
            String messageId = message.getMessageProperties().getHeaders().get("spring_returned_message_correlation").toString();
            if (redisService.hasKey("message:" + messageId)) {
                return;
            }
            redisService.set("message:" + messageId, messageId, 10);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

            String deviceCode = jsonObject.getString("deviceCode");
            if(redisService.hasKey("readingThread:" + deviceCode)) {
                redisService.lSet("device:" + deviceCode, jsonObject, 30, TimeUnit.MINUTES);
            } else {
                redisService.lSet("device:" + deviceCode, jsonObject, 30, TimeUnit.MINUTES);
                //根据设备识别码查询通道号
                String channelId = redisService.hget("sys-netty", deviceCode);
                Thread daemonThread = new Thread(() -> {
                    int count = 0;
                    while (true) {
                        try {
                            if(redisService.hasKey("reading_meter:" + deviceCode)) {
                                if(count < 20) {
                                    Thread.sleep(1000);
                                    count ++;
                                } else {
                                    count = 0;
                                    redisService.del("reading_meter:" + deviceCode);
                                }
                            } else {
                                JSONObject readingJson = (JSONObject) redisService.leftPop("device:" + deviceCode);
                                if (StringUtils.isNull(readingJson)) {
                                    redisService.del("readingThread:" + deviceCode);
                                    break;
                                }
                                redisService.set("reading_meter:" + deviceCode, readingJson, 20);
                                //发送采集指令码
                                nettyChannel.writeToClient(channelId, readingJson.getString("receiveStr"), readingJson.getString("mark"));
                                count = 0;
                            }
                        } catch (InterruptedException e) {
                            log.error("运行异常：" + e.getMessage());
                            throw new RuntimeException(e);
                        }
                    }
                });
                redisService.set("readingThread:" + deviceCode, daemonThread.getId());
                // 将线程设置为守护线程
                daemonThread.setDaemon(true);
                daemonThread.start();
            }
        } catch (IOException e) {
            log.error("jsonObject" + jsonObject.toJSONString());
            e.printStackTrace();
        }
    }
}
